// Copyright 2021-2024 Matrix Origin
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//	http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package readutil

import (
	"fmt"
	"math"

	"github.com/matrixorigin/matrixone/pkg/common/mpool"
	"github.com/matrixorigin/matrixone/pkg/container/types"
	"github.com/matrixorigin/matrixone/pkg/container/vector"
	"github.com/matrixorigin/matrixone/pkg/pb/plan"
	"github.com/matrixorigin/matrixone/pkg/sql/plan/function"
)

const (
	RangeLeftOpen = iota + math.MaxInt16
	RangeRightOpen
	RangeBothOpen
)

type BasePKFilter struct {
	Valid bool
	Op    int
	LB    []byte
	UB    []byte
	Vec   *vector.Vector
	Oid   types.T
	// Disjuncts holds OR-ed atomic filters; when non-empty, Op/LB/UB/Vec are ignored.
	Disjuncts []BasePKFilter
}

func (b *BasePKFilter) String() string {
	name := map[int]string{
		function.LESS_EQUAL:     "less_eq",
		function.LESS_THAN:      "less_than",
		function.GREAT_THAN:     "great_than",
		function.GREAT_EQUAL:    "great_eq",
		RangeLeftOpen:           "range_left_open",
		RangeRightOpen:          "range_right_open",
		RangeBothOpen:           "range_both_open",
		function.EQUAL:          "equal",
		function.IN:             "in",
		function.BETWEEN:        "between",
		function.PREFIX_EQ:      "prefix_in",
		function.PREFIX_IN:      "prefix_in",
		function.PREFIX_BETWEEN: "prefix_between",
	}

	vecStr := "nil"
	if b.Vec != nil {
		//vecStr = common.MoVectorToString(b.Vec, b.Vec.Length())
		vecStr = b.Vec.String()
	}

	var lb, ub any
	if !b.Oid.IsFixedLen() {
		lb = string(b.LB)
		ub = string(b.UB)
	} else {
		lb = types.DecodeValue(b.LB, b.Oid)
		ub = types.DecodeValue(b.UB, b.Oid)
	}

	return fmt.Sprintf("valid = %v, op = %s, lb = %v, ub = %v, vec = %v, oid = %s",
		b.Valid, name[b.Op],
		lb, ub, vecStr, b.Oid.String())
}

func ConstructBasePKFilter(
	expr *plan.Expr,
	tblDef *plan.TableDef,
	mp *mpool.MPool,
) (filter BasePKFilter, err error) {
	if expr == nil {
		return
	}

	defer func() {
		if tblDef.Pkey.CompPkeyCol != nil {
			filter.Oid = types.T_varchar
		}
	}()

	switch exprImpl := expr.Expr.(type) {
	case *plan.Expr_F:
		switch name := exprImpl.F.Func.ObjName; name {
		case "and":
			var filters []BasePKFilter
			for idx := range exprImpl.F.Args {
				ff, err := ConstructBasePKFilter(exprImpl.F.Args[idx], tblDef, mp)
				if err != nil {
					return BasePKFilter{}, err
				}
				if ff.Valid {
					filters = append(filters, ff)
				}
			}

			if len(filters) == 0 {
				return BasePKFilter{}, nil
			}

			for idx := 0; idx < len(filters)-1; {
				f1 := &filters[idx]
				f2 := &filters[idx+1]
				ff, err := mergeFilters(f1, f2, function.AND, mp)
				if err != nil {
					return BasePKFilter{}, err
				}

				if !ff.Valid {
					return BasePKFilter{}, nil
				}

				idx++
				filters[idx] = ff
			}

			for idx := 0; idx < len(filters)-1; idx++ {
				if filters[idx].Vec != nil {
					filters[idx].Vec.Free(mp)
				}
			}

			ret := filters[len(filters)-1]
			return ret, nil

		case "or":
			var (
				filters1 []BasePKFilter
				filters2 []BasePKFilter

				cannotMerge    bool
				hasUnsupported bool
			)

			for idx := range exprImpl.F.Args {
				ff, err := ConstructBasePKFilter(exprImpl.F.Args[idx], tblDef, mp)
				if err != nil {
					return BasePKFilter{}, err
				}
				if !ff.Valid {
					hasUnsupported = true
					continue
				}

				filters1 = append(filters1, toDisjuncts(ff)...)
				filters2 = append(filters2, ff)
			}

			if hasUnsupported {
				return BasePKFilter{}, nil
			}

			if len(filters1) == 0 {
				return BasePKFilter{}, nil
			}

			if len(filters1) == 1 {
				return filters1[0], nil
			}

			for idx := 0; idx < len(filters2)-1; {
				f1 := &filters2[idx]
				f2 := &filters2[idx+1]
				ff, err := mergeFilters(f1, f2, function.OR, mp)
				if err != nil {
					return BasePKFilter{}, nil
				}

				if !ff.Valid {
					//return BasePKFilter{}, nil
					cannotMerge = true
					break
				}

				idx++
				filters2[idx] = ff
			}

			if !cannotMerge {
				for idx := 0; idx < len(filters2)-1; idx++ {
					if filters2[idx].Vec != nil {
						filters2[idx].Vec.Free(mp)
					}
				}

				ret := filters2[len(filters2)-1]
				return ret, nil
			}

			filter.Valid = true
			filter.Disjuncts = filters1
			return filter, nil

		case ">=":
			//a >= ?
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, false, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.GREAT_EQUAL
			filter.LB = vals[0]
			filter.Oid = oid

		case "<=":
			//a <= ?
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, false, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.LESS_EQUAL
			filter.LB = vals[0]
			filter.Oid = oid

		case ">":
			//a > ?
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, false, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.GREAT_THAN
			filter.LB = vals[0]
			filter.Oid = oid

		case "<":
			//a < ?
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, false, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.LESS_THAN
			filter.LB = vals[0]
			filter.Oid = oid

		case "=":
			// a = ?
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, false, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.EQUAL
			filter.LB = vals[0]
			filter.Oid = oid

		case "prefix_eq":
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, false, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.PREFIX_EQ
			filter.LB = vals[0]
			filter.Oid = oid

		case "in":
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, true, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.IN
			{
				vec := vector.NewVec(types.T_any.ToType())
				if err = vec.UnmarshalBinary(vals[0]); err != nil {
					return BasePKFilter{}, err
				}
				filter.Vec = vec
			}
			filter.Oid = oid

		case "prefix_in":
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, true, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.PREFIX_IN
			{
				vec := vector.NewVec(types.T_any.ToType())
				if err = vec.UnmarshalBinary(vals[0]); err != nil {
					return BasePKFilter{}, err
				}
				filter.Vec = vec
			}
			filter.Oid = oid

		case "between":
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, false, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.BETWEEN
			filter.LB = vals[0]
			filter.UB = vals[1]
			filter.Oid = oid

		case "prefix_between":
			ok, oid, vals := evalValue(expr, exprImpl, tblDef, false, tblDef.Pkey.PkeyColName)
			if !ok {
				return
			}
			filter.Valid = true
			filter.Op = function.PREFIX_BETWEEN
			filter.LB = vals[0]
			filter.UB = vals[1]
			filter.Oid = oid

		default:
			//panic(name)
		}
	default:
		//panic(plan2.FormatExpr(expr))
	}

	return
}

func toDisjuncts(f BasePKFilter) []BasePKFilter {
	if len(f.Disjuncts) > 0 {
		return f.Disjuncts
	}
	return []BasePKFilter{f}
}
